package boot.consumer;

import boot.beans.Message;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * <b><code>KafkaReceiver</code></b>
 *
 * <p>Description 接收
 *
 * <p><b>Creation Time:</b>2020-08-06-19:14
 *
 * @author hailan
 * @since kafkalearning 0.0.1
 */
@Component
// @Slf4j
public class KafkaReceiver {
  private static final Logger logger = LoggerFactory.getLogger(KafkaReceiver.class);

  @KafkaListener(topics = {"topic-main"})
  public void listen(ConsumerRecord<?, ?> record) {

    Optional<?> kafkaMessage = Optional.ofNullable(record.value());

    if (kafkaMessage.isPresent()) {

      String message = (String) kafkaMessage.get();
      Message mes = null;
      try {
        mes = new ObjectMapper().configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false).readValue(message, Message.class);
      } catch (JsonProcessingException e) {
        logger.error(e.getMessage(), e);
      }

      logger.info("----------------- 消费消息record =" + record);
      logger.info("------------------ 消费消息message =" + mes);
    }
  }
}
